package com.zhang.third.day03;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Random;

/**
 * @title: 每来一条数据就排序输出
 * @author: zhang
 * @date: 2022/4/6 17:02
 */
public class Example12 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SourceFunction<Integer>() {
                    private boolean running = true;
                    private Random random = new Random();

                    @Override
                    public void run(SourceContext<Integer> ctx) throws Exception {
                        while (running) {
                            ctx.collect(random.nextInt(1000));
                            Thread.sleep(1000L);
                        }
                    }

                    @Override
                    public void cancel() {
                        running = false;
                    }
                })
                .keyBy(r -> r % 2)
                .process(new KeyedProcessFunction<Integer, Integer, String>() {
                    private ListState<Integer> historyData;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        historyData = getRuntimeContext().getListState(
                                new ListStateDescriptor<Integer>(
                                        "history",
                                        Types.INT
                                )
                        );
                    }

                    @Override
                    public void processElement(Integer value, KeyedProcessFunction<Integer, Integer, String>.Context ctx, Collector<String> out) throws Exception {
                        // 如果value的key是0,就添加到0对应的列表中
                        // 如果value的key是1,就添加到1对应的列表中
                        historyData.add(value);
                        ArrayList<Integer> arrayList = new ArrayList<>();
                        for (Integer integer : historyData.get()) {
                            arrayList.add(integer);
                        }

                        arrayList.sort(new Comparator<Integer>() {
                            @Override
                            public int compare(Integer o1, Integer o2) {
                                return o1 - o2;
                            }
                        });

                        StringBuilder result = new StringBuilder();
                        if (ctx.getCurrentKey() == 0) {
                            result.append("偶数历史排序结果如下:\n");
                        } else if (ctx.getCurrentKey() == 1) {
                            result.append("奇数历史排序结果如下:\n");
                        }
                        for (int i = 0; i < arrayList.size(); i++) {
                            result.append(arrayList.get(i));
                            if (i < arrayList.size() - 1) {
                                result.append("=>");
                            }
                        }
                        result.append("\n");

                        out.collect(result.toString());

                    }
                })
                .print();

        env.execute();
    }
}
